package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Student {
  def main(args: Array[String]): Unit = {
    /**
      * 统计偏科最严重的前100名学生
      * 1.计算每个学生分数的方差
      * 2.对方差排序，取前100
      * 3.整理数据
      */
    val conf: SparkConf = new SparkConf().setAppName("Student").setMaster("local")

    val sc = new SparkContext(conf)

    //读取分数表
    val scores: RDD[String] = sc.textFile("data/score.txt")

    //读取学生表
    val cource: RDD[String] = sc.textFile("data/cource.txt")

    //由于分数的范围不一样，所以需要对学生分区做归一化

    //先将获取的两组数据转化成kv格式
    val scoRDD: RDD[(String, String)] = scores.map(sco => {
      (sco.split(",")(1),sco)
    })

    val couRDD: RDD[(String, String)] =cource.map(cou =>{
      (cou.split(",")(0),cou)
    })

    //关联学生表和分数表
    val joinRDD: RDD[(String, (String, String))] = scoRDD.join(couRDD)

    val idAndScoreRDD: RDD[(String, Double)] = joinRDD.map{
      case (_:String,(sco:String,cou:String)) =>
        val scoSplit: Array[String] = sco.split(",")
        val id: String = scoSplit(0)

        val score: Double = scoSplit(2).toDouble

        val couScore: Double = cou.split(",")(2).toDouble

        //对分数进行归一化
        (id,score / couScore)
    }
    //按照id进行分组
    val groupByKeyRDD: RDD[(String, Iterable[Double])] = idAndScoreRDD.groupByKey()

    val stdRDD: RDD[(String, Double)] = groupByKeyRDD.map{
      case (id:String,ss:Iterable[Double]) =>

      /**
        * 计算学生分数的标准差
        */
      val scoList: List[Double] = ss.toList

        //计算平均数
      val avgSco: Double =scoList.sum /scoList.size

        //分数减去平均数在平方
       val chaScore: List[Double] = scoList.map(i =>(i -avgSco)*(i-avgSco))

        //方差的分子
      val fz: Double = chaScore.sum

        //计算标准差
      val std: Double = fz / scoList.size

        (id,std)
    }

    //降序排序取前100
    val top100: Array[(String, Double)] = stdRDD.sortBy(_._2,false).take(100)

    //取这些学生的学号
    val ids: Array[String] = top100.map(_._1)

    //根据这些学号取出分数
    val top100Score: RDD[String] = scores.filter(score =>{
      val id: String = score.split(",")(0)
      ids.contains(id)
    })
    top100Score.foreach(println)
  }
}
